<!DOCTYPE html>
<html>
  <head>
  <meta http-equiv="content-type" content="text/html; charset=utf-8">
  <meta content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=0" name="viewport">
  <meta name="description" content="刘清政">
  <meta name="keyword" content="hexo-theme">
  
    <link rel="shortcut icon" href="/css/images/logo.png">
  
  <title>
    
      python/其他/08-Rabbitmq从入门到精通 | Justin-刘清政的博客
    
  </title>
  <link href="//cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.min.css" rel="stylesheet">
  <link href="//cdnjs.cloudflare.com/ajax/libs/nprogress/0.2.0/nprogress.min.css" rel="stylesheet">
  <link href="//cdnjs.cloudflare.com/ajax/libs/highlight.js/9.12.0/styles/tomorrow.min.css" rel="stylesheet">
  
<link rel="stylesheet" href="/css/style.css">

  
    
<link rel="stylesheet" href="/css/plugins/gitment.css">

  
  <script src="//cdnjs.cloudflare.com/ajax/libs/jquery/3.2.1/jquery.min.js"></script>
  <script src="//cdnjs.cloudflare.com/ajax/libs/geopattern/1.2.3/js/geopattern.min.js"></script>
  <script src="//cdnjs.cloudflare.com/ajax/libs/nprogress/0.2.0/nprogress.min.js"></script>
  
    
<script src="/js/qrious.js"></script>

  
  
    
<script src="/js/gitment.js"></script>

  
  

  
<meta name="generator" content="Hexo 4.2.0"></head>
<div class="wechat-share">
  <img src="/css/images/logo.png" />
</div>

  <body>
    <header class="header fixed-header">
  <div class="header-container">
    <a class="home-link" href="/">
      <div class="logo"></div>
      <span>Justin-刘清政的博客</span>
    </a>
    <ul class="right-list">
      
        <li class="list-item">
          
            <a href="/" class="item-link">主页</a>
          
        </li>
      
        <li class="list-item">
          
            <a href="/tags/" class="item-link">标签</a>
          
        </li>
      
        <li class="list-item">
          
            <a href="/archives/" class="item-link">归档</a>
          
        </li>
      
        <li class="list-item">
          
            <a href="/about/" class="item-link">关于我</a>
          
        </li>
      
    </ul>
    <div class="menu">
      <span class="icon-bar"></span>
      <span class="icon-bar"></span>
      <span class="icon-bar"></span>
    </div>
    <div class="menu-mask">
      <ul class="menu-list">
        
          <li class="menu-item">
            
              <a href="/" class="menu-link">主页</a>
            
          </li>
        
          <li class="menu-item">
            
              <a href="/tags/" class="menu-link">标签</a>
            
          </li>
        
          <li class="menu-item">
            
              <a href="/archives/" class="menu-link">归档</a>
            
          </li>
        
          <li class="menu-item">
            
              <a href="/about/" class="menu-link">关于我</a>
            
          </li>
        
      </ul>
    </div>
  </div>
</header>

    <div id="article-banner">
  <h2>python/其他/08-Rabbitmq从入门到精通</h2>



  <p class="post-date">2020-09-06</p>
    <!-- 不蒜子统计 -->
    <span id="busuanzi_container_page_pv" style='display:none' class="">
        <i class="icon-smile icon"></i> 阅读数：<span id="busuanzi_value_page_pv"></span>次
    </span>
  <div class="arrow-down">
    <a href="javascript:;"></a>
  </div>
</div>
<main class="app-body flex-box">
  <!-- Article START -->
  <article class="post-article">
    <section class="markdown-content"><h2 id="一-消息队列介绍"><a href="#一-消息队列介绍" class="headerlink" title="一 消息队列介绍"></a>一 消息队列介绍</h2><h3 id="1-1-介绍"><a href="#1-1-介绍" class="headerlink" title="1.1 介绍"></a>1.1 介绍</h3><p>消息队列就是基础数据结构中的“先进先出”的一种数据机构。想一下，生活中买东西，需要排队，先排的人先买消费，就是典型的“先进先出”</p>
<p><img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdjzw5s3j31000463zi.jpg" alt="image-20200907000210863"></p>
<h3 id="1-2-MQ解决什么问题"><a href="#1-2-MQ解决什么问题" class="headerlink" title="1.2 MQ解决什么问题"></a>1.2 MQ解决什么问题</h3><p>MQ是一直存在，不过随着微服务架构的流行，成了解决微服务之间问题的常用工具。</p>
<h4 id="应用解耦"><a href="#应用解耦" class="headerlink" title="应用解耦"></a>应用解耦</h4><p>以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。</p>
<p>当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中单用户感受不到物流系统的故障。提升系统的可用性</p>
<p><img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdkm1r3qj310y0dqn3m.jpg" alt="image-20200907000246573"></p>
<h4 id="流量消峰"><a href="#流量消峰" class="headerlink" title="流量消峰"></a>流量消峰</h4><p>举个栗子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。</p>
<p>使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这事有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。</p>
<h4 id="消息分发"><a href="#消息分发" class="headerlink" title="消息分发"></a>消息分发</h4><p>多个服务队数据感兴趣，只需要监听同一类消息即可处理。</p>
<img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdlhzx6mj30ju0ccdhf.jpg" alt="image-20200907000338060" style="zoom:50%;" />

<p>例如A产生数据，B对数据感兴趣。如果没有消息的队列A每次处理完需要调用一下B服务。过了一段时间C对数据也感性，A就需要改代码，调用B服务，调用C服务。只要有服务需要，A服务都要改动代码。很不方便。</p>
<img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdm2jd3ij30u80js0wz.jpg" alt="image-20200907000411649" style="zoom:50%;" />

<p>有了消息队列后，A只管发送一次消息，B对消息感兴趣，只需要监听消息。C感兴趣，C也去监听消息。A服务作为基础服务完全不需要有改动</p>
<h4 id="异步消息"><a href="#异步消息" class="headerlink" title="异步消息"></a>异步消息</h4><img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdmxtqc4j30v60g2tci.jpg" alt="image-20200907000500012" style="zoom:50%;" />

<p>有些服务间调用是异步的，例如A调用B，B需要花费很长时间执行，但是A需要知道B什么时候可以执行完，以前一般有两种方式，A过一段时间去调用B的查询api查询。或者A提供一个callback api，B执行完之后调用api通知A服务。这两种方式都不是很优雅</p>
<img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdnc1tpsj30sa0g00w4.jpg" alt="image-20200907000523720" style="zoom:50%;" />

<p>使用消息总线，可以很方便解决这个问题，A调用B服务后，只需要监听B处理完成的消息，当B处理完成后，会发送一条消息给MQ，MQ会将此消息转发给A服务。</p>
<p>这样A服务既不用循环调用B的查询api，也不用提供callback api。同样B服务也不用做这些操作。A服务还能及时的得到异步处理成功的消息</p>
<h3 id="1-3-常见消息队列及比较"><a href="#1-3-常见消息队列及比较" class="headerlink" title="1.3 常见消息队列及比较"></a>1.3 常见消息队列及比较</h3><p><img src="https://tva1.sinaimg.cn/large/007S8ZIlgy1gihdpf4eudj30m10mkabg.jpg" alt="1"></p>
<p>结论:</p>
<p>Kafka在于分布式架构，RabbitMQ基于AMQP协议来实现，RocketMQ/思路来源于kafka，改成了主从结构，在事务性可靠性方面做了优化。广泛来说，电商、金融等对事务性要求很高的，可以考虑RabbitMQ和RocketMQ，对性能要求高的可考虑Kafka</p>
<h2 id="二-Rabbitmq安装"><a href="#二-Rabbitmq安装" class="headerlink" title="二 Rabbitmq安装"></a>二 Rabbitmq安装</h2><p>官网：<a href="https://www.rabbitmq.com/getstarted.html" target="_blank" rel="noopener">https://www.rabbitmq.com/getstarted.html</a></p>
<h3 id="2-1-服务端原生安装"><a href="#2-1-服务端原生安装" class="headerlink" title="2.1 服务端原生安装"></a>2.1 服务端原生安装</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># 安装配置epel源</span></span><br><span class="line"><span class="comment"># 安装erlang</span></span><br><span class="line">yum -y install erlang</span><br><span class="line"><span class="comment"># 安装RabbitMQ</span></span><br><span class="line">yum -y install rabbitmq-server</span><br></pre></td></tr></table></figure>

<h3 id="2-2-服务端Docker安装"><a href="#2-2-服务端Docker安装" class="headerlink" title="2.2 服务端Docker安装"></a>2.2 服务端Docker安装</h3><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">docker pull rabbitmq:management</span><br><span class="line">docker run -di --name Myrabbitmq -e RABBITMQ_DEFAULT_USER&#x3D;admin -e RABBITMQ_DEFAULT_PASS&#x3D;admin -p 15672:15672 -p 5672:5672 rabbitmq:managemen</span><br></pre></td></tr></table></figure>

<h3 id="2-3-客户端安装"><a href="#2-3-客户端安装" class="headerlink" title="2.3 客户端安装"></a>2.3 客户端安装</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">pip3 install pika</span><br></pre></td></tr></table></figure>

<h3 id="2-4-设置用户和密码"><a href="#2-4-设置用户和密码" class="headerlink" title="2.4 设置用户和密码"></a>2.4 设置用户和密码</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line">rabbitmqctl add_user lqz <span class="number">123</span></span><br><span class="line"><span class="comment"># 设置用户为administrator角色</span></span><br><span class="line">rabbitmqctl set_user_tags lqz administrator</span><br><span class="line"><span class="comment"># 设置权限</span></span><br><span class="line">rabbitmqctl set_permissions -p <span class="string">"/"</span> root <span class="string">".*"</span> <span class="string">".*"</span> <span class="string">".*"</span></span><br><span class="line"></span><br><span class="line"><span class="comment"># 然后重启rabbiMQ服务</span></span><br><span class="line">systemctl reatart rabbitmq-server</span><br><span class="line"> </span><br><span class="line"><span class="comment"># 然后可以使用刚才的用户远程连接rabbitmq server了。</span></span><br></pre></td></tr></table></figure>



<h2 id="三-基于Queue实现生产者消费者模型"><a href="#三-基于Queue实现生产者消费者模型" class="headerlink" title="三 基于Queue实现生产者消费者模型"></a>三 基于Queue实现生产者消费者模型</h2><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> Queue</span><br><span class="line"><span class="keyword">import</span> threading</span><br><span class="line"></span><br><span class="line">message = Queue.Queue(<span class="number">10</span>)</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">producer</span><span class="params">(i)</span>:</span></span><br><span class="line">    <span class="keyword">while</span> <span class="literal">True</span>:</span><br><span class="line">        message.put(i)</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">consumer</span><span class="params">(i)</span>:</span></span><br><span class="line">    <span class="keyword">while</span> <span class="literal">True</span>:</span><br><span class="line">        msg = message.get()</span><br><span class="line"></span><br><span class="line"><span class="keyword">for</span> i <span class="keyword">in</span> range(<span class="number">12</span>):</span><br><span class="line">    t = threading.Thread(target=producer, args=(i,))</span><br><span class="line">    t.start()</span><br><span class="line"></span><br><span class="line"><span class="keyword">for</span> i <span class="keyword">in</span> range(<span class="number">10</span>):</span><br><span class="line">    t = threading.Thread(target=consumer, args=(i,))</span><br><span class="line">    t.start()</span><br></pre></td></tr></table></figure>

<h2 id="四-基本使用（生产者消费者模型）"><a href="#四-基本使用（生产者消费者模型）" class="headerlink" title="四 基本使用（生产者消费者模型）"></a>四 基本使用（生产者消费者模型）</h2><p>对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。</p>
<h3 id="生产者"><a href="#生产者" class="headerlink" title="生产者"></a>生产者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"><span class="comment"># 无密码</span></span><br><span class="line"><span class="comment"># connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))</span></span><br><span class="line"></span><br><span class="line"><span class="comment"># 有密码</span></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列)</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz'</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">''</span>,</span><br><span class="line">                      routing_key=<span class="string">'lqz'</span>, <span class="comment"># 消息队列名称</span></span><br><span class="line">                      body=<span class="string">'hello world'</span>)</span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>

<h3 id="消费者"><a href="#消费者" class="headerlink" title="消费者"></a>消费者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列)</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz'</span>)</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=<span class="string">'lqz'</span>,on_message_callback=callback,auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>

<h2 id="五-消息安全之ack"><a href="#五-消息安全之ack" class="headerlink" title="五 消息安全之ack"></a>五 消息安全之ack</h2><h3 id="生产者-1"><a href="#生产者-1" class="headerlink" title="生产者"></a>生产者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"><span class="comment"># 无密码</span></span><br><span class="line"><span class="comment"># connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))</span></span><br><span class="line"></span><br><span class="line"><span class="comment"># 有密码</span></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列)</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz'</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">''</span>,</span><br><span class="line">                      routing_key=<span class="string">'lqz'</span>, <span class="comment"># 消息队列名称</span></span><br><span class="line">                      body=<span class="string">'hello world'</span>)</span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>

<h3 id="消费者-1"><a href="#消费者-1" class="headerlink" title="消费者"></a>消费者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列)</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz'</span>)</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line">    <span class="comment"># 通知服务端，消息取走了，如果auto_ack=False，不加下面，消息会一直存在</span></span><br><span class="line">    <span class="comment"># ch.basic_ack(delivery_tag=method.delivery_tag)</span></span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=<span class="string">'lqz'</span>,on_message_callback=callback,auto_ack=<span class="literal">False</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>



<h2 id="六-消息安全之durable持久化"><a href="#六-消息安全之durable持久化" class="headerlink" title="六 消息安全之durable持久化"></a>六 消息安全之durable持久化</h2><h3 id="生产者-2"><a href="#生产者-2" class="headerlink" title="生产者"></a>生产者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"><span class="comment"># 无密码</span></span><br><span class="line"><span class="comment"># connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))</span></span><br><span class="line"></span><br><span class="line"><span class="comment"># 有密码</span></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列),durable=True支持持久化，队列必须是新的才可以</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz1'</span>,durable=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">''</span>,</span><br><span class="line">                      routing_key=<span class="string">'lqz1'</span>, <span class="comment"># 消息队列名称</span></span><br><span class="line">                      body=<span class="string">'111'</span>,</span><br><span class="line">                      properties=pika.BasicProperties(</span><br><span class="line">                          delivery_mode=<span class="number">2</span>,  <span class="comment"># make message persistent,消息也持久化</span></span><br><span class="line">                      )</span><br><span class="line">                      )</span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>

<h3 id="消费者-2"><a href="#消费者-2" class="headerlink" title="消费者"></a>消费者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列)</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz1'</span>)</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line">    <span class="comment"># 通知服务端，消息取走了，如果auto_ack=False，不加下面，消息会一直存在</span></span><br><span class="line">    <span class="comment"># ch.basic_ack(delivery_tag=method.delivery_tag)</span></span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=<span class="string">'lqz1'</span>,on_message_callback=callback,auto_ack=<span class="literal">False</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>

<h2 id="七-闲置消费"><a href="#七-闲置消费" class="headerlink" title="七 闲置消费"></a>七 闲置消费</h2><p>正常情况如果有多个消费者，是按照顺序第一个消息给第一个消费者，第二个消息给第二个消费者</p>
<p>但是可能第一个消息的消费者处理消息很耗时，一直没结束，就可以让第二个消费者优先获得闲置的消息</p>
<h3 id="生产者-3"><a href="#生产者-3" class="headerlink" title="生产者"></a>生产者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"><span class="comment"># 无密码</span></span><br><span class="line"><span class="comment"># connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))</span></span><br><span class="line"></span><br><span class="line"><span class="comment"># 有密码</span></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列),durable=True支持持久化，队列必须是新的才可以</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'lqz123'</span>,durable=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">''</span>,</span><br><span class="line">                      routing_key=<span class="string">'lqz123'</span>, <span class="comment"># 消息队列名称</span></span><br><span class="line">                      body=<span class="string">'111'</span>,</span><br><span class="line">                      properties=pika.BasicProperties(</span><br><span class="line">                          delivery_mode=<span class="number">2</span>,  <span class="comment"># make message persistent,消息也持久化</span></span><br><span class="line">                      )</span><br><span class="line">                      )</span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>



<h3 id="消费者-3"><a href="#消费者-3" class="headerlink" title="消费者"></a>消费者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># 声明一个队列(创建一个队列)</span></span><br><span class="line"><span class="comment"># channel.queue_declare(queue='lqz123')</span></span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line">    <span class="comment"># 通知服务端，消息取走了，如果auto_ack=False，不加下面，消息会一直存在</span></span><br><span class="line">    ch.basic_ack(delivery_tag=method.delivery_tag)</span><br><span class="line"></span><br><span class="line">channel.basic_qos(prefetch_count=<span class="number">1</span>) <span class="comment">#####就只有这一句话 谁闲置谁获取，没必要按照顺序一个一个来</span></span><br><span class="line">channel.basic_consume(queue=<span class="string">'lqz123'</span>,on_message_callback=callback,auto_ack=<span class="literal">False</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>



<h2 id="八-发布订阅"><a href="#八-发布订阅" class="headerlink" title="八 发布订阅"></a>八 发布订阅</h2><h3 id="发布者"><a href="#发布者" class="headerlink" title="发布者"></a>发布者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m1'</span>,exchange_type=<span class="string">'fanout'</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">'m1'</span>,</span><br><span class="line">                      routing_key=<span class="string">''</span>,</span><br><span class="line">                      body=<span class="string">'lqz nb'</span>)</span><br><span class="line"></span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>

<h3 id="订阅者-启动几次订阅者会生成几个队列"><a href="#订阅者-启动几次订阅者会生成几个队列" class="headerlink" title="订阅者(启动几次订阅者会生成几个队列)"></a>订阅者(启动几次订阅者会生成几个队列)</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># exchange='m1',exchange(秘书)的名称</span></span><br><span class="line"><span class="comment"># exchange_type='fanout' , 秘书工作方式将消息发送给所有的队列</span></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m1'</span>,exchange_type=<span class="string">'fanout'</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment"># 随机生成一个队列</span></span><br><span class="line">result = channel.queue_declare(queue=<span class="string">''</span>,exclusive=<span class="literal">True</span>)</span><br><span class="line">queue_name = result.method.queue</span><br><span class="line">print(queue_name)</span><br><span class="line"><span class="comment"># 让exchange和queque进行绑定.</span></span><br><span class="line">channel.queue_bind(exchange=<span class="string">'m1'</span>,queue=queue_name)</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>



<h2 id="九-发布订阅高级之Routing-按关键字匹配"><a href="#九-发布订阅高级之Routing-按关键字匹配" class="headerlink" title="九 发布订阅高级之Routing(按关键字匹配)"></a>九 发布订阅高级之Routing(按关键字匹配)</h2><h3 id="发布者-1"><a href="#发布者-1" class="headerlink" title="发布者"></a>发布者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"></span><br><span class="line"><span class="keyword">import</span> pika</span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m2'</span>,exchange_type=<span class="string">'direct'</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">'m2'</span>,</span><br><span class="line">                      routing_key=<span class="string">'bnb'</span>, <span class="comment"># 多个关键字，指定routing_key</span></span><br><span class="line">                      body=<span class="string">'lqz nb'</span>)</span><br><span class="line"></span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>

<h3 id="订阅者1"><a href="#订阅者1" class="headerlink" title="订阅者1"></a>订阅者1</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># exchange='m1',exchange(秘书)的名称</span></span><br><span class="line"><span class="comment"># exchange_type='direct' , 秘书工作方式将消息发送给不同的关键字</span></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m2'</span>,exchange_type=<span class="string">'direct'</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment"># 随机生成一个队列</span></span><br><span class="line">result = channel.queue_declare(queue=<span class="string">''</span>,exclusive=<span class="literal">True</span>)</span><br><span class="line">queue_name = result.method.queue</span><br><span class="line">print(queue_name)</span><br><span class="line"><span class="comment"># 让exchange和queque进行绑定.</span></span><br><span class="line">channel.queue_bind(exchange=<span class="string">'m2'</span>,queue=queue_name,routing_key=<span class="string">'nb'</span>)</span><br><span class="line">channel.queue_bind(exchange=<span class="string">'m2'</span>,queue=queue_name,routing_key=<span class="string">'bnb'</span>)</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>

<h3 id="订阅者2"><a href="#订阅者2" class="headerlink" title="订阅者2"></a>订阅者2</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># exchange='m1',exchange(秘书)的名称</span></span><br><span class="line"><span class="comment"># exchange_type='direct' , 秘书工作方式将消息发送给不同的关键字</span></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m2'</span>,exchange_type=<span class="string">'direct'</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment"># 随机生成一个队列</span></span><br><span class="line">result = channel.queue_declare(queue=<span class="string">''</span>,exclusive=<span class="literal">True</span>)</span><br><span class="line">queue_name = result.method.queue</span><br><span class="line">print(queue_name)</span><br><span class="line"><span class="comment"># 让exchange和queque进行绑定.</span></span><br><span class="line">channel.queue_bind(exchange=<span class="string">'m2'</span>,queue=queue_name,routing_key=<span class="string">'nb'</span>)</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>



<h2 id="九-发布订阅高级之Topic-按关键字模糊匹配"><a href="#九-发布订阅高级之Topic-按关键字模糊匹配" class="headerlink" title="九 发布订阅高级之Topic(按关键字模糊匹配)"></a>九 发布订阅高级之Topic(按关键字模糊匹配)</h2><h3 id="发布者-2"><a href="#发布者-2" class="headerlink" title="发布者"></a>发布者</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m3'</span>,exchange_type=<span class="string">'topic'</span>)</span><br><span class="line"></span><br><span class="line">channel.basic_publish(exchange=<span class="string">'m3'</span>,</span><br><span class="line">                      <span class="comment"># routing_key='lqz.handsome', #都能收到</span></span><br><span class="line">                      routing_key=<span class="string">'lqz.handsome.xx'</span>, <span class="comment">#只有lqz.#能收到</span></span><br><span class="line">                      body=<span class="string">'lqz nb'</span>)</span><br><span class="line"></span><br><span class="line">connection.close()</span><br></pre></td></tr></table></figure>

<h3 id="订阅者1-1"><a href="#订阅者1-1" class="headerlink" title="订阅者1"></a>订阅者1</h3><p>*只能加一个单词</p>
<p>#可以加任意单词字符</p>
<figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># exchange='m1',exchange(秘书)的名称</span></span><br><span class="line"><span class="comment"># exchange_type='direct' , 秘书工作方式将消息发送给不同的关键字</span></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m3'</span>,exchange_type=<span class="string">'topic'</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment"># 随机生成一个队列</span></span><br><span class="line">result = channel.queue_declare(queue=<span class="string">''</span>,exclusive=<span class="literal">True</span>)</span><br><span class="line">queue_name = result.method.queue</span><br><span class="line">print(queue_name)</span><br><span class="line"><span class="comment"># 让exchange和queque进行绑定.</span></span><br><span class="line">channel.queue_bind(exchange=<span class="string">'m3'</span>,queue=queue_name,routing_key=<span class="string">'lqz.#'</span>)</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>

<h3 id="订阅者2-1"><a href="#订阅者2-1" class="headerlink" title="订阅者2"></a>订阅者2</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"></span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># exchange='m1',exchange(秘书)的名称</span></span><br><span class="line"><span class="comment"># exchange_type='topic' , 模糊匹配</span></span><br><span class="line">channel.exchange_declare(exchange=<span class="string">'m3'</span>,exchange_type=<span class="string">'topic'</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment"># 随机生成一个队列</span></span><br><span class="line">result = channel.queue_declare(queue=<span class="string">''</span>,exclusive=<span class="literal">True</span>)</span><br><span class="line">queue_name = result.method.queue</span><br><span class="line">print(queue_name)</span><br><span class="line"><span class="comment"># 让exchange和queque进行绑定.</span></span><br><span class="line">channel.queue_bind(exchange=<span class="string">'m3'</span>,queue=queue_name,routing_key=<span class="string">'lqz.*'</span>)</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">callback</span><span class="params">(ch, method, properties, body)</span>:</span></span><br><span class="line">  	queue_name = result.method.queue <span class="comment"># 发送的routing_key是什么</span></span><br><span class="line">    print(<span class="string">"消费者接受到了任务: %r"</span> % body)</span><br><span class="line"></span><br><span class="line">channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>



<h2 id="十-基于rabbitmq实现rpc"><a href="#十-基于rabbitmq实现rpc" class="headerlink" title="十 基于rabbitmq实现rpc"></a>十 基于rabbitmq实现rpc</h2><h3 id="服务端"><a href="#服务端" class="headerlink" title="服务端"></a>服务端</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line">credentials = pika.PlainCredentials(<span class="string">"admin"</span>,<span class="string">"admin"</span>)</span><br><span class="line">connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>,credentials=credentials))</span><br><span class="line">channel = connection.channel()</span><br><span class="line"></span><br><span class="line"><span class="comment"># 起翰监听任务队列</span></span><br><span class="line">channel.queue_declare(queue=<span class="string">'rpc_queue'</span>)</span><br><span class="line"></span><br><span class="line"><span class="function"><span class="keyword">def</span> <span class="title">on_request</span><span class="params">(ch, method, props, body)</span>:</span></span><br><span class="line">    n = int(body)</span><br><span class="line">    response = n + <span class="number">100</span></span><br><span class="line">    <span class="comment"># props.reply_to  要放结果的队列.</span></span><br><span class="line">    <span class="comment"># props.correlation_id  任务</span></span><br><span class="line">    ch.basic_publish(exchange=<span class="string">''</span>,</span><br><span class="line">                     routing_key=props.reply_to,</span><br><span class="line">                     properties=pika.BasicProperties(correlation_id= props.correlation_id),</span><br><span class="line">                     body=str(response))</span><br><span class="line">    ch.basic_ack(delivery_tag=method.delivery_tag)</span><br><span class="line"></span><br><span class="line">channel.basic_qos(prefetch_count=<span class="number">1</span>)</span><br><span class="line">channel.basic_consume( queue=<span class="string">'rpc_queue'</span>,on_message_callback=on_request,)</span><br><span class="line">channel.start_consuming()</span><br></pre></td></tr></table></figure>

<h3 id="客户端"><a href="#客户端" class="headerlink" title="客户端"></a>客户端</h3><figure class="highlight python"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> pika</span><br><span class="line"><span class="keyword">import</span> uuid</span><br><span class="line"></span><br><span class="line"><span class="class"><span class="keyword">class</span> <span class="title">FibonacciRpcClient</span><span class="params">(object)</span>:</span></span><br><span class="line">    <span class="function"><span class="keyword">def</span> <span class="title">__init__</span><span class="params">(self)</span>:</span></span><br><span class="line">        credentials = pika.PlainCredentials(<span class="string">"admin"</span>, <span class="string">"admin"</span>)</span><br><span class="line">        self.connection = pika.BlockingConnection(pika.ConnectionParameters(<span class="string">'101.133.225.166'</span>, credentials=credentials))</span><br><span class="line">        self.channel = self.connection.channel()</span><br><span class="line"></span><br><span class="line">        <span class="comment"># 随机生成一个消息队列(用于接收结果)</span></span><br><span class="line">        result = self.channel.queue_declare(queue=<span class="string">''</span>,exclusive=<span class="literal">True</span>)</span><br><span class="line">        self.callback_queue = result.method.queue</span><br><span class="line"></span><br><span class="line">        <span class="comment"># 监听消息队列中是否有值返回,如果有值则执行 on_response 函数(一旦有结果,则执行on_response)</span></span><br><span class="line">        self.channel.basic_consume(queue=self.callback_queue,on_message_callback=self.on_response, auto_ack=<span class="literal">True</span>)</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">def</span> <span class="title">on_response</span><span class="params">(self, ch, method, props, body)</span>:</span></span><br><span class="line">        <span class="keyword">if</span> self.corr_id == props.correlation_id:</span><br><span class="line">            self.response = body</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">def</span> <span class="title">call</span><span class="params">(self, n)</span>:</span></span><br><span class="line">        self.response = <span class="literal">None</span></span><br><span class="line">        self.corr_id = str(uuid.uuid4())</span><br><span class="line"></span><br><span class="line">        <span class="comment"># 客户端 给 服务端 发送一个任务:  任务id = corr_id / 任务内容 = '30' / 用于接收结果的队列名称</span></span><br><span class="line">        self.channel.basic_publish(exchange=<span class="string">''</span>,</span><br><span class="line">                                   routing_key=<span class="string">'rpc_queue'</span>, <span class="comment"># 服务端接收任务的队列名称</span></span><br><span class="line">                                   properties=pika.BasicProperties(</span><br><span class="line">                                         reply_to = self.callback_queue, <span class="comment"># 用于接收结果的队列</span></span><br><span class="line">                                         correlation_id = self.corr_id, <span class="comment"># 任务ID</span></span><br><span class="line">                                         ),</span><br><span class="line">                                   body=str(n))</span><br><span class="line">        <span class="keyword">while</span> self.response <span class="keyword">is</span> <span class="literal">None</span>:</span><br><span class="line">            self.connection.process_data_events()</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> self.response</span><br><span class="line"></span><br><span class="line">fibonacci_rpc = FibonacciRpcClient()</span><br><span class="line"></span><br><span class="line">response = fibonacci_rpc.call(<span class="number">50</span>)</span><br><span class="line">print(<span class="string">'返回结果:'</span>,response)</span><br></pre></td></tr></table></figure>



</section>
    <!-- Tags START -->
    
    <!-- Tags END -->
    <!-- NAV START -->
    
  <div class="nav-container">
    <!-- reverse left and right to put prev and next in a more logic postition -->
    
      <a class="nav-left" href="/python/%E5%85%B6%E4%BB%96/07-ZeroRPC%E5%92%8CSimpleXMLRPCServer/">
        <span class="nav-arrow">← </span>
        
          python/其他/07-ZeroRPC和SimpleXMLRPCServer
        
      </a>
    
    
      <a class="nav-right" href="/go/4-Go%E8%AF%AD%E8%A8%80%E6%A1%86%E6%9E%B6/4-1-gin%E6%A1%86%E6%9E%B6/1-gin%E6%A1%86%E6%9E%B6%E7%AE%80%E4%BB%8B/">
        
          go/4-Go语言框架/4-1-gin框架/1-gin框架简介
        
        <span class="nav-arrow"> →</span>
      </a>
    
  </div>

    <!-- NAV END -->
    <!-- 打赏 START -->
    
      <div class="money-like">
        <div class="reward-btn">
          赏
          <span class="money-code">
            <span class="alipay-code">
              <div class="code-image"></div>
              <b>使用支付宝打赏</b>
            </span>
            <span class="wechat-code">
              <div class="code-image"></div>
              <b>使用微信打赏</b>
            </span>
          </span>
        </div>
        <p class="notice">点击上方按钮,请我喝杯咖啡！</p>
      </div>
    
    <!-- 打赏 END -->
    <!-- 二维码 START -->
    
      <div class="qrcode">
        <canvas id="share-qrcode"></canvas>
        <p class="notice">扫描二维码，分享此文章</p>
      </div>
    
    <!-- 二维码 END -->
    
      <!-- Gitment START -->
      <div id="comments"></div>
      <!-- Gitment END -->
    
  </article>
  <!-- Article END -->
  <!-- Catalog START -->
  
    <aside class="catalog-container">
  <div class="toc-main">
  <!-- 不蒜子统计 -->
    <strong class="toc-title">目录</strong>
    
      <ol class="toc-nav"><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#一-消息队列介绍"><span class="toc-nav-text">一 消息队列介绍</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#1-1-介绍"><span class="toc-nav-text">1.1 介绍</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#1-2-MQ解决什么问题"><span class="toc-nav-text">1.2 MQ解决什么问题</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-4"><a class="toc-nav-link" href="#应用解耦"><span class="toc-nav-text">应用解耦</span></a></li><li class="toc-nav-item toc-nav-level-4"><a class="toc-nav-link" href="#流量消峰"><span class="toc-nav-text">流量消峰</span></a></li><li class="toc-nav-item toc-nav-level-4"><a class="toc-nav-link" href="#消息分发"><span class="toc-nav-text">消息分发</span></a></li><li class="toc-nav-item toc-nav-level-4"><a class="toc-nav-link" href="#异步消息"><span class="toc-nav-text">异步消息</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#1-3-常见消息队列及比较"><span class="toc-nav-text">1.3 常见消息队列及比较</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#二-Rabbitmq安装"><span class="toc-nav-text">二 Rabbitmq安装</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#2-1-服务端原生安装"><span class="toc-nav-text">2.1 服务端原生安装</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#2-2-服务端Docker安装"><span class="toc-nav-text">2.2 服务端Docker安装</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#2-3-客户端安装"><span class="toc-nav-text">2.3 客户端安装</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#2-4-设置用户和密码"><span class="toc-nav-text">2.4 设置用户和密码</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#三-基于Queue实现生产者消费者模型"><span class="toc-nav-text">三 基于Queue实现生产者消费者模型</span></a></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#四-基本使用（生产者消费者模型）"><span class="toc-nav-text">四 基本使用（生产者消费者模型）</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#生产者"><span class="toc-nav-text">生产者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#消费者"><span class="toc-nav-text">消费者</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#五-消息安全之ack"><span class="toc-nav-text">五 消息安全之ack</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#生产者-1"><span class="toc-nav-text">生产者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#消费者-1"><span class="toc-nav-text">消费者</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#六-消息安全之durable持久化"><span class="toc-nav-text">六 消息安全之durable持久化</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#生产者-2"><span class="toc-nav-text">生产者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#消费者-2"><span class="toc-nav-text">消费者</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#七-闲置消费"><span class="toc-nav-text">七 闲置消费</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#生产者-3"><span class="toc-nav-text">生产者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#消费者-3"><span class="toc-nav-text">消费者</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#八-发布订阅"><span class="toc-nav-text">八 发布订阅</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#发布者"><span class="toc-nav-text">发布者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#订阅者-启动几次订阅者会生成几个队列"><span class="toc-nav-text">订阅者(启动几次订阅者会生成几个队列)</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#九-发布订阅高级之Routing-按关键字匹配"><span class="toc-nav-text">九 发布订阅高级之Routing(按关键字匹配)</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#发布者-1"><span class="toc-nav-text">发布者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#订阅者1"><span class="toc-nav-text">订阅者1</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#订阅者2"><span class="toc-nav-text">订阅者2</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#九-发布订阅高级之Topic-按关键字模糊匹配"><span class="toc-nav-text">九 发布订阅高级之Topic(按关键字模糊匹配)</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#发布者-2"><span class="toc-nav-text">发布者</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#订阅者1-1"><span class="toc-nav-text">订阅者1</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#订阅者2-1"><span class="toc-nav-text">订阅者2</span></a></li></ol></li><li class="toc-nav-item toc-nav-level-2"><a class="toc-nav-link" href="#十-基于rabbitmq实现rpc"><span class="toc-nav-text">十 基于rabbitmq实现rpc</span></a><ol class="toc-nav-child"><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#服务端"><span class="toc-nav-text">服务端</span></a></li><li class="toc-nav-item toc-nav-level-3"><a class="toc-nav-link" href="#客户端"><span class="toc-nav-text">客户端</span></a></li></ol></li></ol>
    
  </div>
</aside>
  
  <!-- Catalog END -->
</main>

<script>
  (function () {
    var url = 'http://www.liuqingzheng.top/python/其他/08-Rabbitmq从入门到精通/';
    var banner = ''
    if (banner !== '' && banner !== 'undefined' && banner !== 'null') {
      $('#article-banner').css({
        'background-image': 'url(' + banner + ')'
      })
    } else {
      $('#article-banner').geopattern(url)
    }
    $('.header').removeClass('fixed-header')

    // error image
    $(".markdown-content img").on('error', function() {
      $(this).attr('src', 'http://file.muyutech.com/error-img.png')
      $(this).css({
        'cursor': 'default'
      })
    })

    // zoom image
    $(".markdown-content img").on('click', function() {
      var src = $(this).attr('src')
      if (src !== 'http://file.muyutech.com/error-img.png') {
        var imageW = $(this).width()
        var imageH = $(this).height()

        var zoom = ($(window).width() * 0.95 / imageW).toFixed(2)
        zoom = zoom < 1 ? 1 : zoom
        zoom = zoom > 2 ? 2 : zoom
        var transY = (($(window).height() - imageH) / 2).toFixed(2)

        $('body').append('<div class="image-view-wrap"><div class="image-view-inner"><img src="'+ src +'" /></div></div>')
        $('.image-view-wrap').addClass('wrap-active')
        $('.image-view-wrap img').css({
          'width': `${imageW}`,
          'transform': `translate3d(0, ${transY}px, 0) scale3d(${zoom}, ${zoom}, 1)`
        })
        $('html').css('overflow', 'hidden')

        $('.image-view-wrap').on('click', function() {
          $(this).remove()
          $('html').attr('style', '')
        })
      }
    })
  })();
</script>


  <script>
    var qr = new QRious({
      element: document.getElementById('share-qrcode'),
      value: document.location.href
    });
  </script>



  <script>
    var gitmentConfig = "liuqingzheng";
    if (gitmentConfig !== 'undefined') {
      var gitment = new Gitment({
        id: "python/其他/08-Rabbitmq从入门到精通",
        owner: "liuqingzheng",
        repo: "FuckBlog",
        oauth: {
          client_id: "32a4076431cf39d0ecea",
          client_secret: "94484bd79b3346a949acb2fda3c8a76ce16990c6"
        },
        theme: {
          render(state, instance) {
            const container = document.createElement('div')
            container.lang = "en-US"
            container.className = 'gitment-container gitment-root-container'
            container.appendChild(instance.renderHeader(state, instance))
            container.appendChild(instance.renderEditor(state, instance))
            container.appendChild(instance.renderComments(state, instance))
            container.appendChild(instance.renderFooter(state, instance))
            return container;
          }
        }
      })
      gitment.render(document.getElementById('comments'))
    }
  </script>




    <div class="scroll-top">
  <span class="arrow-icon"></span>
</div>
    <footer class="app-footer">
<!-- 不蒜子统计 -->
<span id="busuanzi_container_site_pv">
     本站总访问量<span id="busuanzi_value_site_pv"></span>次
</span>
<span class="post-meta-divider">|</span>
<span id="busuanzi_container_site_uv" style='display:none'>
     本站访客数<span id="busuanzi_value_site_uv"></span>人
</span>
<script async src="//busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js"></script>



  <p class="copyright">
    &copy; 2021 | Proudly powered by <a href="https://www.cnblogs.com/xiaoyuanqujing" target="_blank">小猿取经</a>
    <br>
    Theme by <a href="https://www.cnblogs.com/xiaoyuanqujing" target="_blank" rel="noopener">小猿取经</a>
  </p>
</footer>

<script>
  function async(u, c) {
    var d = document, t = 'script',
      o = d.createElement(t),
      s = d.getElementsByTagName(t)[0];
    o.src = u;
    if (c) { o.addEventListener('load', function (e) { c(null, e); }, false); }
    s.parentNode.insertBefore(o, s);
  }
</script>
<script>
  async("//cdnjs.cloudflare.com/ajax/libs/fastclick/1.0.6/fastclick.min.js", function(){
    FastClick.attach(document.body);
  })
</script>

<script>
  var hasLine = 'true';
  async("//cdnjs.cloudflare.com/ajax/libs/highlight.js/9.12.0/highlight.min.js", function(){
    $('figure pre').each(function(i, block) {
      var figure = $(this).parents('figure');
      if (hasLine === 'false') {
        figure.find('.gutter').hide();
      }
      var lang = figure.attr('class').split(' ')[1] || 'code';
      var codeHtml = $(this).html();
      var codeTag = document.createElement('code');
      codeTag.className = lang;
      codeTag.innerHTML = codeHtml;
      $(this).attr('class', '').empty().html(codeTag);
      figure.attr('data-lang', lang.toUpperCase());
      hljs.highlightBlock(block);
    });
  })
</script>





<!-- Baidu Tongji -->

<script>
    var _baId = 'c5fd96eee1193585be191f318c3fa725';
    // Originial
    var _hmt = _hmt || [];
    (function() {
      var hm = document.createElement("script");
      hm.src = "//hm.baidu.com/hm.js?" + _baId;
      var s = document.getElementsByTagName("script")[0];
      s.parentNode.insertBefore(hm, s);
    })();
</script>


<script src="/js/script.js"></script>


<script src="/js/search.js"></script>


<script src="/js/load.js"></script>



  <span class="local-search local-search-google local-search-plugin" style="right: 50px;top: 70px;;position:absolute;z-index:2;">
      <input type="search" placeholder="站内搜索" id="local-search-input" class="local-search-input-cls" style="">
      <div id="local-search-result" class="local-search-result-cls"></div>
  </span>


  </body>
</html>